package com.miaocat19.mq;

import com.alibaba.fastjson.JSON;
import com.miaocat19.dao.StockLogDOMapper;
import com.miaocat19.dataobject.StockLogDO;
import com.miaocat19.error.BusinessException;
import com.miaocat19.service.OrderService;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

@Component
public class MqProducer {

    private DefaultMQProducer producer;

    private TransactionMQProducer transactionMQProducer;

    @Value("${mq.nameserver.addr}")
    private String nameAddr;

    @Value("${mq.topicname}")
    private String topicName;

    @Autowired
    private OrderService orderService;

    @Autowired
    private StockLogDOMapper stockLogDOMapper;

    @PostConstruct
    public void init() throws MQClientException {
        producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr(nameAddr);
        producer.start();


        transactionMQProducer = new TransactionMQProducer("transaction_producer_group");
        transactionMQProducer.setNamesrvAddr(nameAddr);
        transactionMQProducer.start();

        transactionMQProducer.setTransactionListener(new TransactionListener() {

            /*这个是我们真正要做的事   我们要在这个中进行创建订单  这个我们就是通过org 拿来对象*/
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
                Integer itemId = (Integer) ((Map) arg).get("itemId");
                Integer promoId = (Integer) ((Map) arg).get("promoId");
                Integer userId = (Integer) ((Map) arg).get("userId");
                Integer amount = (Integer) ((Map) arg).get("amount");
                String stockLogId = (String) ((Map) arg).get("stockLogId");
                try {
                    orderService.createOrder(userId, itemId, promoId, amount, stockLogId);
                } catch (BusinessException e) {
                    /*这个地方我们创建订单报错之后       我们就要更新 日志表中的状态 把他创建为     订单三表示回滚*/
                    e.printStackTrace();
                    StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId);
                    stockLogDO.setItemId(3);
                    stockLogDOMapper.updateByPrimaryKeySelective(stockLogDO);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            /*长时间不知道executeLocalTransaction方法中返回的时候什么状态 我们就定时去查看这个状态  我们就拿message 来进行检查
            * 这个地方是在executeLocalTransaction  执行发生等待时没有明确的返回成功或者是unkon状态 我们执行下面的方法  我么这个方法会去查看当前的方法到底执行到什么时候了  因为上面的执行的时候
            * 设置了  库存日志对应的值     因此 我们设置相应的值  进行操作来查看对应的库存日志  做出显影的回应
            * */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //我们是根据是否扣减库存，来判断要返回的commit.roolback还是继续unkonow
                String jsonString = new String(msg.getBody());
                Map<String, Object> map = JSON.parseObject(jsonString, Map.class);
                Integer itemId = (Integer) map.get("itemId");
                Integer amount = (Integer) map.get("amount");
                String stockLogId = (String) map.get("stockLogId");

                StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId);

                if (stockLogDO == null) {
                    return LocalTransactionState.UNKNOW;
                }

                if (stockLogDO.getStatus().intValue() == 2) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (stockLogDO.getStatus().intValue() == 1) {
                    return LocalTransactionState.UNKNOW;
                }

                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

    }


    public boolean transactionasyncReduceStock(Integer userId, Integer itemId, Integer promoId, Integer amount, String stockLogId) {

        Map<String, Object> map = new HashMap<>();
        map.put("itemId", itemId);
        map.put("amount", amount);
        map.put("stockLogId", stockLogId);


        Map<String, Object> argsmap = new HashMap<>();

        argsmap.put("itemId", itemId);
        argsmap.put("amount", amount);
        argsmap.put("userId", userId);
        argsmap.put("promoId", promoId);
        argsmap.put("stockLogId", stockLogId);

        Message message = new Message(topicName, "increase", JSON.toJSON(map).toString().getBytes(Charset.forName("UTF-8")));

        TransactionSendResult sendResult = null;

        try {
            /*我们执行这个方法之后 我们需要先去执行相应的  executeLocalTransaction 因为这个方法下面的这个方法是prepare的 是没有提交的 */
            sendResult = transactionMQProducer.sendMessageInTransaction(message, argsmap);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        }

        if (sendResult.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
            return false;
        } else if (sendResult.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) {
            return true;
        } else {
            return false;
        }
    }


    /*同步扣减操作*/
    public boolean asyncReduceStock(Integer itemId, Integer amount) {
        Map<String, Object> map = new HashMap<>();

        map.put("itemId", itemId);
        map.put("amount", amount);

        Message message = new Message(topicName, "increase", JSON.toJSON(map).toString().getBytes(Charset.forName("UTF-8")));
        try {
            producer.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        } catch (RemotingException e) {
            e.printStackTrace();
            return false;
        } catch (MQBrokerException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
}
